package zklock

import (
	"errors"
	"fmt"
	"github.com/google/uuid"
	"github.com/samuel/go-zookeeper/zk"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"
)

//Lock 基于Zookeeper的分布式锁
type Lock struct {
	clientLock   sync.RWMutex //本地锁 用于控制加锁和解锁的语义
	prefix       string       //锁前缀 比如 lock_
	lockFullPath string       //锁在zk中的实际绝对路径 比如 /locks/lock-0000001
	rootPath     string       //锁在zk中的根节点路径 比如 /locks ，相关的锁都会创建在该路径下 这个根节点是持久节点
	seq          int64        //锁的顺序节点的自增序号
	nodeACL      []zk.ACL     //锁节点的权限
	conn         *zk.Conn     //实际的zk客户端连接
}

func NewLock(conn *zk.Conn, rootPath string, acl ...zk.ACL) (*Lock, error) {
	if conn == nil {
		return nil, errors.New("必须指定conn")
	}
	if strings.TrimSpace(rootPath) == "" {
		return nil, errors.New("必须指定有效的rootPath")
	}
	if !strings.HasPrefix(rootPath, "/") {
		return nil, errors.New("必须指定有效的rootPath")
	}
	var locker Lock
	locker.rootPath = rootPath
	locker.prefix = "lock_" //目前固定为lock_
	locker.nodeACL = zk.WorldACL(zk.PermAll)
	if len(acl) > 0 {
		locker.nodeACL = acl
	}
	locker.conn = conn
	exists, _, err := conn.Exists(rootPath)
	if err != nil {
		return nil, fmt.Errorf("检测%s是否存在发生错误：%w", rootPath, err)
	}
	//如果根路径不存在则创建 一般根路径是创建好的 否者权限不好确定；根路径最好根据业务区分
	if !exists {
		if _, err = conn.Create(rootPath, []byte("lock base"), 0, zk.WorldACL(zk.PermAll)); err != nil {
			return nil, fmt.Errorf("创建%s失败：%w", rootPath, err)
		}
	}
	return &locker, nil
}

//LockBlock 阻塞直到获取锁成功
func (z *Lock) LockBlock() {
	//zk的分布式锁依赖临时顺序节点的序号来判断是否已获取到锁
	//若这里不额外加锁限制，则未UnLock前重复调用Lock可能会导致lockPath被覆盖而死锁
	z.clientLock.Lock()
	z.block()
}

//UnlockBlock 阻塞直到释放锁成功
func (z *Lock) UnlockBlock() {
	defer z.clientLock.Unlock()

	if strings.TrimSpace(z.lockFullPath) == "" {
		return
	}
	//不考虑不正确使用Lock和UnLock的情况，这种属于代码错误

	for {
		err := z.conn.Delete(z.lockFullPath, -1)
		//删除无错误或节点不存在算作成功
		if err == nil || err == zk.ErrNoNode {
			z.lockFullPath = ""
			break
		}
		time.Sleep(100 * time.Millisecond)
	}
}

func (z *Lock) createNode() (string, error) {
	//在 rootPath 下 创建 临时有序 子节点
	lockId := uuid.New().String()
	//如： /locks/lock_xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx_
	pathPrefix := z.rootPath + "/" + z.prefix + lockId + "_"
	createTime := time.Now().Format("2006-01-02 15:04:05.000")
	nodePath, err := z.conn.Create(pathPrefix, []byte(createTime), zk.FlagEphemeral|zk.FlagSequence, z.nodeACL)
	if err != nil {
		return lockId, ErrCreateNode
	}
	//如：/locks/lock_xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx_0000000001
	z.lockFullPath = nodePath
	z.seq = z.getNodeSeq(strings.TrimPrefix(nodePath, z.rootPath+"/")) // 如 1
	return lockId, nil
}

func (z *Lock) create() bool {
	if z.lockFullPath != "" {
		return true
	}
	curLockId, err := z.createNode()
	if err == nil {
		return true
	}
	if err != ErrCreateNode {
		return false
	}
	//创建临时节点失败可能只是应答超时，若忽略而重复创建，则之前的节点无法释放导致死锁
	//因此这里失败时尝试获取一下根节点下的子节点，判断是否存在刚刚创建的节点，若不存在才是真的创建失败
	var nodes []node
	for {
		nodes, err = z.getValidSortedNodes()
		if err == nil {
			break
		}
		//获取子节点失败则重试 直到成功
		if err == ErrGetChildren {
			time.Sleep(500 * time.Millisecond)
			continue
		}
		return false
	}
	found := false
	for _, v := range nodes {
		if v.id == curLockId {
			found = true
			z.lockFullPath = v.fullPath
			z.seq = v.seq
			break
		}
	}
	return found
}

func (z *Lock) block() {
	for {
		//1.创建节点
		if !z.create() {
			time.Sleep(1 * time.Second)
			continue
		}
		//2.判断锁节点是否是所有子节点中序号最小的节点  若是则代表获取锁成功
		isMin, prevNodeFullPath, err := z.checkMinSeqNode()
		if err != nil {
			if err == ErrNoCurNode {
				//只有匹配不到当前节点才需要重新创建
				z.lockFullPath = ""
			}
			continue
		}
		if isMin {
			break
		}

		//3.检查前一个序号的节点是否存在
		// 若存在则watch等待其删除，否者重新判断是否是最小节点
		//这里Exists同时watch，保证原子性，否者Exists和Watch之间若该节点已删除，则永远收不到该节点删除的事件
		exists, _, watch, err := z.conn.ExistsW(prevNodeFullPath)
		if err != nil || !exists {
			continue
		}

		//等待前一个节点删除 无论是否是删除事件 都重新检查当前节点是否是最小节点
		<-watch
	}

}

type node struct {
	fullPath string
	seq      int64
	id       string
}

func (z *Lock) getValidSortedNodes() ([]node, error) {
	children, _, err := z.conn.Children(z.rootPath)
	if err != nil {
		return nil, ErrGetChildren
	}
	var validChildren []string
	for _, v := range children {
		if !strings.HasPrefix(v, z.prefix) {
			continue
		}
		validChildren = append(validChildren, v)
	}
	var childNodes []node
	for _, path := range validChildren {
		seq := z.getNodeSeq(path)
		if seq == -1 {
			continue
		}
		id := strings.TrimPrefix(path, z.prefix)
		id = id[:36]
		childNodes = append(childNodes, node{
			fullPath: z.rootPath + "/" + path,
			seq:      seq,
			id:       id,
		})
	}

	if len(childNodes) == 0 {
		return nil, ErrNoValidChildren
	}

	sort.Slice(childNodes, func(i, j int) bool {
		return childNodes[i].seq < childNodes[j].seq
	})

	return childNodes, nil
}

func (z *Lock) checkMinSeqNode() (bool, string, error) {
	//2.创建成功 判断当前节点是否是所有子节点中序号最小的，若是则获取锁成功
	childNodes, err := z.getValidSortedNodes()
	if err != nil {
		return false, "", err
	}
	curNodeIndex := -1
	for i, n := range childNodes {
		if n.seq == z.seq {
			curNodeIndex = i
			break
		}
	}
	if curNodeIndex == 0 {
		return true, "", nil
	}
	if curNodeIndex == -1 {
		return false, "", ErrNoCurNode
	}
	return false, childNodes[curNodeIndex-1].fullPath, nil
}

func (z *Lock) getNodeSeq(nodePath string) int64 {
	//注意nodePath不包含basePath部分，如：lock-0000000001
	if !strings.HasPrefix(nodePath, z.prefix) {
		return -1
	}
	//获取节点序号字符串 如：0000000001
	seqStr := strings.TrimPrefix(nodePath, z.prefix)
	seqStr = seqStr[37:]
	ret, err := strconv.ParseInt(seqStr, 10, 64)
	if err != nil {
		return -1
	}
	return ret
}

var (
	ErrGetChildren     = errors.New("获取子节点失败")
	ErrNoValidChildren = errors.New("无有效子节点")
	ErrNoCurNode       = errors.New("无法匹配到当前节点")
	ErrCreateNode      = errors.New("创建节点失败")
)
